package com.sui.bigdata.flink.sql.side;

import org.junit.Test;

/**
 * @Author: dongguosheng
 * @Date: 2019/9/4 19:50
 * @Review
 * @Email: guosheng_dong@sui.com
 */
public class MailListConcentration extends demo {
    @Test
    public void testMailListConcentration() throws Exception {

        String sql1 = "create aggregate function PhoneHHI with com.feidee.flink.sql.function.aggregate.PhoneHHI ; \n" +
                "create scalar function str2md5 with com.feidee.flink.sql.function.scalar.Md5; \n" +
                "create scalar function AESDecrypt with com.feidee.flink.sql.function.scalar.AESDecrypt;\n" +
                "create aggregate function UserknPhoneIspNum with com.feidee.flink.sql.function.aggregate.UserknPhoneIspNum;\n" +
                "create aggregate function UserknPhoneAll with com.feidee.flink.sql.function.aggregate.UserknPhoneAll;\n" +
                "create aggregate function UserknPhoneUnusualNum with com.feidee.flink.sql.function.aggregate.UserknPhoneUnusualNum;\n" +
                "create aggregate function UserknNameUnusualNum with com.feidee.flink.sql.function.aggregate.UserknNameUnusualNum;\n" +
                "\n" +
                "CREATE TABLE MailList (\n" +
                "    suid VARCHAR,\n" +
                "    name VARCHAR,\n" +
                "    public_phone VARCHAR\n" +
                ") WITH (\n" +
                "    type = 'kafka',\n" +
                "    bootstrapServers = '10.201.7.187:9093,10.201.7.188:9093',\n" +
                "    offsetReset = 'latest',\n" +
                "    topic = 'etl_dwb.sswl.contact_info_phone_suid',\n" +
                "    groupId = 'nb02',\n" +
                "    parallelism = '1'\n" +
                ");\n" +
                "\n" +
                "CREATE TABLE ResultTable (\n" +
                "    info:suidmd5 VARCHAR,\n" +
                "    info:suid VARCHAR,\n" +
                "    info:userkn_phone_head_5_hhi DOUBLE,\n" +
                "    info:userkn_phone_head_6_hhi DOUBLE,\n" +
                "    info:userkn_phone_head_7_hhi DOUBLE,\n" +
                "    info:userkn_phone_head_8_hhi DOUBLE,\n" +
                "    info:userkn_phone_head_9_hhi DOUBLE,\n" +
                "    info:userkn_phone_4_4_hhi DOUBLE,\n" +
                "    info:userkn_phone_7_4_hh DOUBLE,\n" +
                "    info:userkn_phone_isp_cnt int,\n" +
                "    info:userkn_phone_all int,\n" +
                "    info:userkn_phone_unusual_cnt int,\n" +
                "    info:userkn_name_unusual_cnt int\n" +
                ") WITH (\n" +
                "    type = 'hbase',\n" +
                "    zookeeperQuorum = '10.201.7.114:2181,10.201.7.115:2181,10.201.7.113:2181',\n" +
                "    zookeeperParent = '/hbase',\n" +
                "    tableName = 'adl_bd_ls_loan_credit_wallet',\n" +
                "    rowkey = 'info:suidmd5'\n" +
                ");\n" +
                "CREATE TABLE phone_test (\n" +
                "                id varchar,\n" +
                "                phone_pre varchar,\n" +
                "                PRIMARY KEY(phone_pre),\n" +
                "                PERIOD FOR SYSTEM_TIME\n" +
                "                 ) WITH (\n" +
                "                 type='mysql',\n" +
                "                 url='jdbc:mysql://tidb1.testfeideedba.com:4000/dt?useUnicode=true&amp;characterEncoding=utf8;useSSL=true',\n" +
                "                 userName='DSJ_user',\n" +
                "                 password='606cde3507b026H,dHb_77e67H_e',\n" +
                "                 tableName='test_phone');\n" +
                "    \n" +
//                "INSERT INTO ResultTable " +
                "    SELECT\n" +
                "    CONCAT(md5(a.suid), 'creditwallet') AS suidmd5,\n" +
                "    a.suid,\n" +
                "    PhoneHHI(AESDecrypt(a.public_phone), 0, 5) AS userkn_phone_head_5_hhi,\n" +
                "    PhoneHHI(AESDecrypt(a.public_phone), 0, 6) AS userkn_phone_head_6_hhi,\n" +
                "    PhoneHHI(AESDecrypt(a.public_phone), 0, 7) AS userkn_phone_head_7_hhi,\n" +
                "    PhoneHHI(AESDecrypt(a.public_phone), 0, 8) AS userkn_phone_head_8_hhi,\n" +
                "    PhoneHHI(AESDecrypt(a.public_phone), 0, 9) AS userkn_phone_head_9_hhi,\n" +
                "    PhoneHHI(AESDecrypt(a.public_phone), 4) AS userkn_phone_4_4_hhi,\n" +
                "    PhoneHHI(AESDecrypt(a.public_phone), -4) AS userkn_phone_7_4_hhi,\n" +
                "    UserknPhoneIspNum (b.id,AESDecrypt(a.public_phone)) AS userkn_phone_isp_cnt,\n" +
                "    UserknPhoneAll(AESDecrypt(a.public_phone)) AS userkn_phone_all,\n" +
                "    UserknPhoneUnusualNum(AESDecrypt(a.public_phone)) AS userkn_phone_unusual_cnt,\n" +
                "    UserknNameUnusualNum(AESDecrypt(a.name)) AS userkn_name_unusual_cnt\n" +
                "FROM\n" +
                "    (\n" +
                "        SELECT\n" +
                "            proctime,\n" +
                "            public_phone,\n" +
                "            substring(AESDecrypt(public_phone),1,3) AS prephone,\n" +
                "            suid,\n" +
                "            name\n" +
                "        FROM\n" +
                "            MailList\n" +
                "    ) a\n" +
                "LEFT JOIN phone_test b ON a.prephone = b.phone_pre\n" +
                "GROUP BY\n" +
                "    SESSION (\n" +
                "        a.proctime,\n" +
                "        INTERVAL '60' SECOND\n" +
                "    ),\n" +
                "    a.suid;";
        test2(sql1);
    }
}
